package cn.simon.free.redis.mq.core.consumer.task;

import cn.simon.free.redis.mq.core.consumer.AbstractRedisMessageConsumer;
import cn.simon.free.redis.mq.enums.ConsumerType;
import cn.simon.free.redis.mq.model.Message;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.extern.slf4j.Slf4j;

import java.util.Optional;

import static jodd.util.ThreadUtil.sleep;

/**
 * @author simon
 * @date 2021/7/2 10:08
 * @copyright:
 * @modifier
 * @motto a no talent, people who have to study hard
 */
@Slf4j
@Builder
@AllArgsConstructor
public class LongPollingTask implements Runnable {

    protected final AbstractRedisMessageConsumer longPollingConsumer;
    protected final String consumerQueue;
    protected final ConsumerType consumerType;

    @Override
    public void run() {
        boolean isTerminal = true;
        while (isTerminal) {
            try {
                Message message = longPollingConsumer.getMessage(consumerQueue, consumerType);
                if (Optional.ofNullable(message).isPresent()) {
                    consume(message);
                } else {
                    sleep(1000);
                }
            } catch (Exception e) {
                log.error("redis message consumer long pool error,queue:{}", consumerQueue, e);
                isTerminal = false;
            }
        }
    }

    /**
     * 调用子类,并根据返回值删除消息
     *
     * @param message
     */
    private void consume(Message message) {
        if (longPollingConsumer.processMessage(consumerQueue, message, consumerType)) {
            longPollingConsumer.deleteMessage(consumerQueue, message);
        }
    }

}
